package com.ctsi.djcloudmqtt.mq;

import com.ctsi.djcloudmqtt.config.MqttProperties;
import com.ctsi.djcloudmqtt.dto.WeatherDTO;
import com.ctsi.djcloudmqtt.dto.down.DJParentDTO;
import com.ctsi.djcloudmqtt.dto.up.osb.OsbDTO;
import com.ctsi.djcloudmqtt.service.MqttService;
import com.ctsi.djcloudmqtt.util.CommonConstants;
import com.ctsi.djcloudmqtt.util.json.jackson.JacksonUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.locks.LockSupport;

@Component
@Slf4j
public class MqttAcceptCallback implements MqttCallbackExtended {


    public static   volatile WeatherDTO weatherDTO = new WeatherDTO();


    @Resource
    private MqttAcceptClient    mqttAcceptClient;

    @Resource
    private MqttProperties mqttProperties;

    @Value("${topicList.thing.product.servicesReply}")
    private String thingProductServicesReply;

    @Value("${topicList.thing.product.gatewayOsd}")
    private String thingProductGatewayOsd;

    @Value("${topicList.thing.product.state}")
    private String thingProductGatewayState;

    @Value("${topicList.thing.product.events}")
    private String thingProductGatewayEvents;

    @Value("${topicList.gatewaySn}")
    private String gatewaySn;

    @Value("${topicList.deviceSn}")
    private String deviceSn;

    /**
     * 客户端断开后触发
     *
     * @param throwable
     */
    @Override
    public void connectionLost(Throwable throwable) {
        log.info("连接断开，可以重连");
        if (MqttAcceptClient.client == null || !MqttAcceptClient.client.isConnected()) {
            log.info("【emqx重新连接】....................................................");
            mqttAcceptClient.reconnection();
        }
    }

    /**
     * 客户端收到消息触发
     *
     * @param topic       主题
     * @param mqttMessage 消息
     */
    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        log.info("【接收消息主题】:" + topic);
        log.info("【接收消息Qos】:" + mqttMessage.getQos());

        String message = new String(mqttMessage.getPayload());

        log.info("【接收消息内容】:" + message);
        try {

            ObjectMapper mapper = new ObjectMapper();

            thingProductGatewayOsd  = thingProductGatewayOsd.replace(MqttService.gatewaySnKeyWord,gatewaySn);
            if(topic.equals(thingProductGatewayOsd)){
                OsbDTO osbDTO = JacksonUtil.string2Bean(message,OsbDTO.class);
                if(null != osbDTO && null != osbDTO.getData()){
                    if(null != osbDTO.getData().getRainfall()){
                        weatherDTO.setRainFall(osbDTO.getData().getRainfall());
                    }
                   if(null != osbDTO.getData().getWindSpeed()){
                    weatherDTO.setWindSpeed(osbDTO.getData().getWindSpeed());
                    }
                    if(null != osbDTO.getData().getEnvironmentTemperature()){
                        weatherDTO.setEnvironmentTemperature(osbDTO.getData().getEnvironmentTemperature());
                    }
                    if(null != osbDTO.getData().getHumidity()) {
                        weatherDTO.setHumidity(osbDTO.getData().getHumidity());
                    }
                }
                return;
            }

            JsonNode rootNode = mapper.readTree(message);
            JsonNode bidNameNode = rootNode.get(DJParentDTO.Fields.bid);
            if(null != bidNameNode){
                String bid = bidNameNode.asText();
                Thread thread = MqttService.threadCache.getIfPresent(bid);
                if(null != thread){
                    MqttService.jsonCache.put(bid,message);
                    log.info("监测到有线程阻塞，激活该线程："+message);
                    LockSupport.unpark(thread);
                    return;
                }

                JsonNode methodNameNode = rootNode.get(DJParentDTO.Fields.method);
                if(null != methodNameNode){
                    String methodName = methodNameNode.asText();
                    if(CommonConstants.DJServiceMethod.TAKEOFF_2_POINT_PROGRESS.equals(methodName)){
                        MqttService.processCache.put(bid+CommonConstants.DJServiceMethod.TAKEOFF_2_POINT_PROGRESS,rootNode.get(DJParentDTO.Fields.data).asText());
                    }
                    if(CommonConstants.DJServiceMethod.SPEAKER_TTS_PLAY_START_PROGRESS.equals(methodName)){
                        MqttService.processCache.put(bid+CommonConstants.DJServiceMethod.SPEAKER_TTS_PLAY_START_PROGRESS,rootNode.get(DJParentDTO.Fields.data).asText());
                    }

                    if(CommonConstants.DJServiceMethod.FLY_2_POINT_PROGRESS.equals(methodName)){
                        MqttService.processCache.put(bid+CommonConstants.DJServiceMethod.FLY_2_POINT_PROGRESS,rootNode.get(DJParentDTO.Fields.data).asText());
                    }
                }
            }
        }catch (JsonProcessingException e){
            log.info("报错:",e);
        }
    }

    /**
     * 发布消息成功
     *
     * @param token token
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        String[] topics = token.getTopics();
        for (String topic : topics) {
            log.info("向主题【" + topic + "】发送消息成功！");
        }
        try {
            MqttMessage message = token.getMessage();
            byte[] payload = message.getPayload();
            String s = new String(payload, StandardCharsets.UTF_8);
            log.info("【消息内容】:" + s);
        } catch (Exception e) {
            log.info("MqttAcceptCallback deliveryComplete error,message:{}", e.getMessage());
            log.info("报错",e);
        }
    }

    /**
     * 连接emq服务器后触发
     *
     * @param b
     * @param s
     */
    @Override
    public void connectComplete(boolean b, String s) {
        log.info("============================= 客户端【" + MqttAcceptClient.client.getClientId() + "】连接成功！=============================");

        mqttAcceptClient.subscribe(mqttProperties.getDefaultTopic(), 0);
        log.info("============================= 客户端监听主题："+mqttProperties.getDefaultTopic()+"成功");

        thingProductServicesReply  = thingProductServicesReply.replace(MqttService.gatewaySnKeyWord,gatewaySn);
        mqttAcceptClient.subscribe(thingProductServicesReply, 0);
        log.info("============================= 客户端监听主题："+thingProductServicesReply+"成功");

        thingProductGatewayOsd  = thingProductGatewayOsd.replace(MqttService.gatewaySnKeyWord,gatewaySn);
        mqttAcceptClient.subscribe(thingProductGatewayOsd, 0);
        log.info("============================= 客户端监听主题："+thingProductGatewayOsd+"成功");

        thingProductGatewayState  = thingProductGatewayState.replace(MqttService.gatewaySnKeyWord,gatewaySn);
        mqttAcceptClient.subscribe(thingProductGatewayState, 0);
        log.info("============================= 客户端监听主题："+thingProductGatewayState+"成功");

        thingProductGatewayEvents  = thingProductGatewayEvents.replace(MqttService.gatewaySnKeyWord,gatewaySn);
        mqttAcceptClient.subscribe(thingProductGatewayEvents, 0);
        log.info("============================= 客户端监听主题："+thingProductGatewayEvents+"成功");
    }
}